package com.example.rabbitmqdemo.mq;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.Arrays;
import java.util.Objects;

/**
 * @author liujb
 * @description TODO 消费者
 * @date 2022-06-28 22:44
 */
public class Consumer {

    static class TestA{

        public static void main(String[] args) {
//            workQueues("TestA");
//            pubAndSubFanout("TestA");
//            pubAndSubDirect("error");
            pubAndSubTopic("user.*");
        }
    }
    static class TestB{
        public static void main(String[] args) {
//            workQueues("TestB");
//            pubAndSubFanout("TestB");
//            pubAndSubDirect( "info","warning","error");
            pubAndSubTopic("user.#");

        }

    }



    public static void main(String[] args) {
        // 第一种队列模型
//        helloWorld();
        // 第二种队列模型：
        // 我们开启两个线程执行消息消费
//        workQueues("");
        // 发布者/订阅者 - Fanout
//        pubAndSubFanout("");
        // 发布者/订阅者 - Direct
//        pubAndSubDirect("info");
        // 发布者/订阅者 - Topic
//        pubAndSubTopic("");
    }

    private static void pubAndSubTopic(String routingKey) {
        try{
            // 获取连接
            Connection connection = RabbitmqUtils.getConnection();
            // 获取通道
            if (Objects.nonNull(connection)) {
                Channel channel = connection.createChannel();
                // 声明交换机与交换机类型
                channel.exchangeDeclare("topics", BuiltinExchangeType.TOPIC);
                // 生成临时队列
                String queue = channel.queueDeclare().getQueue();
                // 绑定交换机与队列 使用通配符形式的routingKey
                channel.queueBind(queue, "topics", routingKey);

                channel.basicConsume(queue, true, new DefaultConsumer(channel) {
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                        System.out.println(new String(body));
                    }
                });
            }
        }catch (IOException e){
            e.printStackTrace();
        }
    }

    private static void pubAndSubDirect(String ...names) {
        try{
            // 获取连接
            Connection connection = RabbitmqUtils.getConnection();
            if (Objects.nonNull(connection)) {
                // 获取通道
                Channel channel = connection.createChannel();
                String exchangeName = "test_direct_exchange";
                String str = Arrays.toString(names).replace("[", "")
                        .replace("]", "").replace(", ", "_");
                String queueName = "test_direct_queue_" + str;
                // 声明交换机
                channel.exchangeDeclare(exchangeName,BuiltinExchangeType.DIRECT);
                // 建立绑定关系
                channel.queueDeclare(queueName, false, false, false, null);
                // 由于交换机指定类型为Direct 在声明队列时需要指定 routing Key
                for (int i = 0; i < names.length; i++) {
                    channel.queueBind(queueName,exchangeName,names[i]);
                }
                // 消费消息
                channel.basicConsume(queueName, false, new DefaultConsumer(channel) {
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                        System.out.println(new String(body));
                    }
                });
            }
        }catch (IOException e){
            e.printStackTrace();
        }
    }

    private static void pubAndSubFanout(String name) {
        try{
            // 获取连接
            Connection connection = RabbitmqUtils.getConnection();
            if(Objects.nonNull(connection)){
                // 获取通道
                Channel channel = connection.createChannel();
                // 声明交换机
                channel.exchangeDeclare("logs", "fanout");
                // 临时队列
                String queue = channel.queueDeclare().getQueue();
                // 绑定交换机和队列
                //  var1：队列名称
                //  var2：交换机名称
                //  var3：路由键
                channel.queueBind(queue, "logs", "");
                // 消费消息
                channel.basicConsume(queue,true,new DefaultConsumer(channel){
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                        System.out.println(name + ":" + new String(body)); // 打印消息
                    }
                });

            }

        }catch (IOException e){
            e.printStackTrace();
        }

    }


    // 第二种队列模型
    private static void workQueues(String name) {
        Connection connection = null;
        Channel channel = null;
        try {
            // 获取连接
            connection = RabbitmqUtils.getConnection();
            if (Objects.nonNull(connection )) {
                // 获取通道
                channel = connection.createChannel();
                String queue = "work";
                // 声明队列
                channel.queueDeclare(queue, true, false, false, null);

                channel.basicQos(1); // 设置每次只发送一条消息给同一消费者
                // 消费消息
                Channel finalChannel = channel;
                // 参数2：autoAck-如果服务器应考虑消息一旦传递就已确认，则为true; 如果服务器应期望显式确认，则为false
                channel.basicConsume(queue,false,new DefaultConsumer(finalChannel){
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                        if(Objects.equals(name,"TestA")){
                            try {
                                Thread.sleep(1000); // 每次TestA消费消息前睡眠1秒
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                            }
                        }
                        System.out.println(name + "===>" + new String(body)); // 打印消息
                        // 开启消息手动确认
                        //  参数1：队列中消息确认消息标识
                        //  参数2：是否开启多个消息同时确
                        finalChannel.basicAck(envelope.getDeliveryTag(),false);
                    }
                });
            }
        }catch (IOException e){
            e.printStackTrace();
        }
    }


    /* 第一种：Hello World! */
    private static void helloWorld() {
        Connection connection = null;
        Channel channel = null;

        try {
            // 获取连接对象
            connection = RabbitmqUtils.getConnection();
            // 获取通道
            if (Objects.nonNull(connection)) {
                channel = connection.createChannel();
                String queue = "hello";
                /*
                通道绑定对应消息队列
                     参数1 queue：队列名称(不存在自动创建)
                     参数2 durable：用来定义队列特性是否需要持久化(为true该队列将在服务器重启后保留下来)
                     参数3 exclusive：是否独占队列(为true仅限此连接)
                     参数4 autoDelete：是否在消费完成不再使用后自动删除队列
                     参数5 arguments：队列的其他属性（构造参数）
                 */
                channel.queueDeclare(queue, false, false, false, null);
                // 消费消息
                DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
                    // 获取消息并处理。此方法类似于事件监听，有消息时会自动调用
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                        System.out.println("message：" + new String(body)); // body即消息体
                    }
                };
                /*
                    参数1 queue：队列名称
                    参数2 autoAck：开启消息的自动确认机制
                    参数3 Consumer callback：消费时的回调接口
                 */
                channel.basicConsume(queue, true, defaultConsumer);

            }

        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            System.out.println("消费者消费消息完成......");
        }
    }



}
